package com.cskaoyan.order.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.order.mq.properties.CreatPromoOrderConsumerProperties;
import com.cskaoyan.order.service.impl.OrderPromoServiceImpl;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @ClassName CreatPromoOrderConsumer
 * @Description TODO
 * @Author XiaoWJ
 * @Date 2022/9/29 21:53
 **/
@Component
public class CreatPromoOrderConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    CreatPromoOrderConsumerProperties properties;

    @Autowired
    OrderPromoServiceImpl orderPromoService;

    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer(properties.getId());

        consumer.setNamesrvAddr(properties.getAddr());

        consumer.setMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) ->
                {
                    // 获取消息
                    MessageExt message = list.get(0);

                    byte[] body = message.getBody();

                    String obj = null;
                    try {
                        obj = new String(body, 0, body.length, properties.getCharset());
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }

                    CreateSeckillOrderRequest seckillOrderRequest = JSON.parseObject(obj, CreateSeckillOrderRequest.class);

                    // 调用service层的插入订单服务
                    try {
                        orderPromoService.createPromoOrder(seckillOrderRequest);

                    }catch (Exception e){
                        e.printStackTrace();

                        // 出现异常，服务消费失败
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    // 服务消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                );

        // 开启消费服务
        try {
            // 订阅主题
            consumer.subscribe(properties.getTopic(),"*");
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }


    }

}
